package com.strp.mptcp.path;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.Data;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Subflow：表示一个独立的 TCP 子流
 * - 负责建立和维护与服务器的连接
 * - 处理数据的发送和接收
 * - 维护连接状态和统计信息
 * - 实现心跳机制
 *
 * @author 曹行阳
 * @date 2025/6/12 11:29
 * @version 0.1
 */
@Data
public class Subflow {
    private final int pathId;
    private final SocketAddress remoteAddress;
    private final Bootstrap bootstrap;
    private final SslContext sslContext;
    private final int maxReceiveQueueSize;
    private final int readTimeout;
    private final int writeTimeout;
    private final int connectionTimeout;
    private final int maxRetries;
    private Channel channel;
    private final AtomicBoolean active = new AtomicBoolean(false);
    private final AtomicLong lastHeartbeatTime = new AtomicLong(System.currentTimeMillis());

    // 心跳配置
    private static final int HEARTBEAT_INTERVAL_SECONDS = 5;
    private static final int HEARTBEAT_TIMEOUT_SECONDS = 15;
    private static final long RECONNECT_DELAY_MS = 1000;

    private int reconnectAttempts = 0;
    private volatile boolean reconnecting = false;

    public Subflow(int pathId, SocketAddress remoteAddress, Bootstrap bootstrap, SslContext sslContext,
                   int maxReceiveQueueSize, int readTimeout, int writeTimeout,
                   int connectionTimeout, int maxRetries) {
        this.pathId = pathId;
        this.remoteAddress = remoteAddress;
        this.bootstrap = bootstrap;
        this.sslContext = sslContext;
        this.maxReceiveQueueSize = maxReceiveQueueSize;
        this.readTimeout = readTimeout;
        this.writeTimeout = writeTimeout;
        this.connectionTimeout = connectionTimeout;
        this.maxRetries = maxRetries;
    }

    public void connect() {
        if (reconnecting) {
            return;
        }

        // 配置此子流的 pipeline
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) {
                ChannelPipeline p = ch.pipeline();

                // 添加 SSL 处理器
                if (sslContext != null) {
                    p.addLast(sslContext.newHandler(ch.alloc(), getHostName(), getPort()));
                }

                // 添加空闲检测
                p.addLast(new IdleStateHandler(readTimeout, writeTimeout, HEARTBEAT_INTERVAL_SECONDS));

                // 添加业务处理器
                p.addLast(new SubflowHandler(Subflow.this));
            }
        });

        ChannelFuture future = bootstrap.connect(remoteAddress);
        future.addListener((ChannelFutureListener) f -> {
            if (f.isSuccess()) {
                this.channel = f.channel();
                this.active.set(true);
                this.reconnectAttempts = 0;
                this.reconnecting = false;
                System.out.println("Subflow " + pathId + " connected to " + remoteAddress);
            } else {
                handleConnectionFailure(f.cause());
            }
        });
    }

    private void handleConnectionFailure(Throwable cause) {
        System.err.println("Subflow " + pathId + " failed to connect: " + cause.getMessage());
        this.active.set(false);

        if (reconnectAttempts < maxRetries) {
            reconnectAttempts++;
            reconnecting = true;
            bootstrap.config().group().schedule(this::connect, RECONNECT_DELAY_MS, TimeUnit.MILLISECONDS);
        } else {
            System.err.println("Subflow " + pathId + " failed to reconnect after " + maxRetries + " attempts");
        }
    }

    public void send(ByteBuf data) {
        if (isActive()) {
            channel.writeAndFlush(data).addListener((ChannelFutureListener) future -> {
                if (!future.isSuccess()) {
                    System.err.println("Failed to send data on subflow " + pathId + ": " + future.cause());
                    handleSendFailure();
                }
            });
        } else {
            System.err.println("Cannot send data on inactive subflow " + pathId);
        }
    }

    private void handleSendFailure() {
        if (isActive()) {
            close();
            if (reconnectAttempts < maxRetries) {
                reconnectAttempts++;
                reconnecting = true;
                channel.eventLoop().schedule(this::connect, RECONNECT_DELAY_MS, TimeUnit.MILLISECONDS);
            }
        }
    }

    public boolean isActive() {
        return active.get() && channel != null && channel.isActive();
    }

    public void close() {
        active.set(false);
        if (channel != null) {
            channel.close();
        }
    }

    public void startHeartbeat() {
        if (channel != null && channel.isActive()) {
            channel.pipeline().addLast(new HeartbeatHandler(this));
        }
    }

    private String getHostName() {
        if (remoteAddress instanceof InetSocketAddress) {
            return ((InetSocketAddress) remoteAddress).getHostName();
        }
        return "unknown";
    }

    private int getPort() {
        if (remoteAddress instanceof InetSocketAddress) {
            return ((InetSocketAddress) remoteAddress).getPort();
        }
        return -1;
    }

    private static class SubflowHandler extends ChannelInboundHandlerAdapter {
        private final Subflow subflow;

        public SubflowHandler(Subflow subflow) {
            this.subflow = subflow;
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof ByteBuf data) {
                subflow.lastHeartbeatTime.set(System.currentTimeMillis());
                System.out.println("Received data on subflow " + subflow.getPathId());
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            System.err.println("Error in SubflowHandler: " + cause.getMessage());
            ctx.close();
            subflow.handleConnectionFailure(cause);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) {
            System.out.println("Subflow " + subflow.getPathId() + " disconnected");
            subflow.active.set(false);
            ctx.fireChannelInactive();
        }
    }

    private static class HeartbeatHandler extends ChannelInboundHandlerAdapter {
        private final Subflow subflow;

        public HeartbeatHandler(Subflow subflow) {
            this.subflow = subflow;
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent event) {
                if (event.state() == IdleState.READER_IDLE) {
                    long currentTime = System.currentTimeMillis();
                    if (currentTime - subflow.lastHeartbeatTime.get() > HEARTBEAT_TIMEOUT_SECONDS * 1000) {
                        System.err.println("Heartbeat timeout on subflow " + subflow.getPathId());
                        ctx.close();
                        subflow.handleConnectionFailure(new Exception("Heartbeat timeout"));
                    }
                } else if (event.state() == IdleState.WRITER_IDLE) {
                    // 发送心跳包
                    ByteBuf heartbeat = Unpooled.buffer(1);
                    heartbeat.writeByte(0);
                    ctx.writeAndFlush(heartbeat);
                }
            }
            ctx.fireUserEventTriggered(evt);
        }
    }
}